package com.gmscosmo.janus.engine;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.gmscosmo.common.JanusNativeSignaling;
import com.gmscosmo.common.JanusPlugin;
import com.gmscosmo.janus.sendMessage.IMSender;
import com.gmscosmo.janus.sendMessage.JanusRequestSender;
import com.gmscosmo.janus.sendMessage.WebSocketSender;
import com.gmscosmo.janus.util.JanusLogUtil;
import com.gmscosmo.janus.util.JanusUtil;
import com.gmscosmo.janus.lock.RedissLockUtil;
import com.gmscosmo.model.*;
import com.gmscosmo.request.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author stephen.qiu
 */
@Component
public class AVEngineKit {
    @Autowired
    JanusUtil janusUtil;
    @Autowired
    JanusLogUtil janusLogUtil;
    @Autowired
    JanusRequestSender janusRequestSender;
    @Autowired
    IMSender imSender;
    @Autowired
    WebSocketSender webSocketSender;

    /**
     * 创建janus session
     *
     * @param userId
     * @return
     * @throws Exception
     */
    public long createSession(String userId) throws Exception {
        JanusRequest createSessionRequest = new JanusRequest(JanusNativeSignaling.CREATE);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(createSessionRequest, nodeId).get();
        JSONObject data = response.getJSONObject("data");
        long session_id = data.getLong("id");
        janusUtil.addEventListener(userId, session_id);
        return session_id;
    }

    /**
     * 添加janus 插件
     *
     * @throws Exception
     */
    public long createHandle(String userId, long session_id, String plugin) throws Exception {
        JanusRequest attachRequest = new JanusRequest(JanusNativeSignaling.ATTACH);
        attachRequest.setPlugin(plugin);
        attachRequest.setSession_id(session_id);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(attachRequest, nodeId).get();
        JSONObject data = response.getJSONObject("data");
        long handle_id = data.getLong("id");
        return handle_id;
    }

    /**
     * 开始拉流
     *
     * @param userId
     * @param targetUserId
     * @param jsep
     * @return
     * @throws Exception
     */
    public String start(String userId, String targetUserId, Jsep jsep) throws Exception {
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandleIds().get(targetUserId);
        String room = janusUtil.getUserRoom(userId);
        JanusRequest startRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        startRequest.setJsep(jsep);
        JanusRequestBody startBody = new JanusRequestBody(JanusNativeSignaling.START);
        startBody.setRoom(room);
        startRequest.setBody(startBody);
        startRequest.setSession_id(session_id);
        startRequest.setHandle_id(handle_id);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(startRequest, nodeId).get();
        if (response.toJSONString().contains("started")) {
            JSONObject plugindata = response.getJSONObject("plugindata");
            JSONObject data = plugindata.getJSONObject("data");
            if (targetUserId != null) {
                data.put("display", targetUserId);
                plugindata.put("data", data);
                response.put("plugindata", plugindata);
            }
        }
        return response.toString();
    }

    /**
     * 创建视频会议转发流
     * @param userId
     * @param targetUserId
     * @param rtpForwardData
     * @return
     * @throws Exception
     */
    public String createStream(String userId, String targetUserId, RtpForwardData rtpForwardData) throws Exception {
        JanusRequest createStreamRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandleIds().get(targetUserId);
        createStreamRequest.setSession_id(session_id);
        createStreamRequest.setHandle_id(handle_id);
        CreateStreamBody createStreamBody = new CreateStreamBody(JanusNativeSignaling.CREATE);
        int streamId = rtpForwardData.getId();
        createStreamBody.setType("rtp");
        createStreamBody.setId(streamId);
        createStreamBody.setAudio(true);
        createStreamBody.setVideo(true);
        createStreamBody.setDescription(userId + "|" + streamId);
        createStreamBody.setAudioport(rtpForwardData.getAudio_port());
        createStreamBody.setVideoport(rtpForwardData.getVideo_port());
        createStreamBody.setVideoport2(rtpForwardData.getVideo_port_2());
        createStreamBody.setVideoport3(rtpForwardData.getVideo_port_3());
        createStreamBody.setVideortcpport(rtpForwardData.getVideo_rtcp_port());
        createStreamBody.setVideosimulcast(true);
        createStreamRequest.setBody(createStreamBody);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(createStreamRequest, nodeId).get();
        addSubscriber(session_id, handle_id, targetUserId, streamId, rtpForwardData, userId);
        return response.toString();
    }

    /**
     * 创建直播转发流
     * @param userId
     * @param rtpForwardData
     * @return
     * @throws Exception
     */
    public String createStream(String userId, RtpForwardData rtpForwardData) throws Exception {
        JanusRequest createStreamRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandleIds().get(userId);
        createStreamRequest.setSession_id(session_id);
        createStreamRequest.setHandle_id(handle_id);
        CreateStreamBody createStreamBody = new CreateStreamBody(JanusNativeSignaling.CREATE);
        int streamId = Integer.parseInt(janusUtil.getUserRoom(userId));
        createStreamBody.setType("rtp");
        createStreamBody.setId(streamId);
        createStreamBody.setAudio(true);
        createStreamBody.setVideo(true);
        createStreamBody.setDescription(userId + "|" + streamId);
        createStreamBody.setAudioport(rtpForwardData.getAudio_port());
        createStreamBody.setVideoport(rtpForwardData.getVideo_port());
        createStreamBody.setVideoport2(rtpForwardData.getVideo_port_2());
        createStreamBody.setVideoport3(rtpForwardData.getVideo_port_3());
        createStreamBody.setVideortcpport(rtpForwardData.getVideo_rtcp_port());
        createStreamBody.setVideosimulcast(true);
        createStreamRequest.setBody(createStreamBody);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(createStreamRequest, nodeId).get();
        addSubscriber(session_id, handle_id, userId, streamId, rtpForwardData, userId);
        return response.toString();
    }

    private void addSubscriber(long session_id, long handle_id, String targetUserId, int streamId, RtpForwardData rtpForwardData, String userId) {
        Subscriber subscriber = new Subscriber();
        subscriber.setSession_id(session_id);
        subscriber.setHandle_id(handle_id);
        subscriber.setAudio_port(rtpForwardData.getAudio_port());
        subscriber.setVideo_port(rtpForwardData.getVideo_port());
        subscriber.setVideo_port_2(rtpForwardData.getVideo_port_2());
        subscriber.setVideo_port_3(rtpForwardData.getVideo_port_3());
        subscriber.setId(streamId);
        subscriber.setSubscriberFrom(targetUserId);
        subscriber.setVideo_rtcp_port(rtpForwardData.getVideo_rtcp_port());
        janusUtil.putSubscribers(userId, subscriber);
    }

    public void destroyMyStream(String userId) {
        JanusRequest destroyStreamRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        DestroyStreamBody destroyStreamBody = new DestroyStreamBody();
        List<Subscriber> subscriberList = janusUtil.getSubscribers(userId);
        for (Subscriber subscriber : subscriberList) {
            destroyStreamRequest.setSession_id(subscriber.getSession_id());
            janusUtil.removePort(subscriber.getAudio_port());
            janusUtil.removePort(subscriber.getVideo_port());
            janusUtil.removePort(subscriber.getVideo_port_2());
            janusUtil.removePort(subscriber.getVideo_port_3());
            if (subscriber.getVideo_rtcp_port() != null) {
                janusUtil.removePort(subscriber.getVideo_rtcp_port());
            }
            destroyStreamBody.setId(subscriber.getId());
            destroyStreamRequest.setHandle_id(subscriber.getHandle_id());
            destroyStreamRequest.setBody(destroyStreamBody);
            int nodeId = janusUtil.getUserNode(userId);
            janusRequestSender.sendWithFuture(destroyStreamRequest, nodeId);
        }
        janusUtil.removeSubscribers(userId);
    }


    public String watch(String userId, String targetUserId, int watchId) throws Exception {
        JanusRequest watchRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        watchRequest.setJanus(JanusNativeSignaling.MESSAGE);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandleIds().get(targetUserId);
        watchRequest.setSession_id(session_id);
        watchRequest.setHandle_id(handle_id);
        WatchBody watchBody = new WatchBody();
        watchBody.setRequest(JanusNativeSignaling.WATCH);
        watchBody.setId(watchId);
        watchRequest.setBody(watchBody);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(watchRequest, nodeId).get();
        return response.toString();
    }


    public boolean roomExist(String userId, int roomId) throws Exception {
        JanusRequest roomExistRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        JanusRequestBody roomExistBody = new JanusRequestBody(JanusNativeSignaling.EXISTS);
        roomExistBody.setRoom(roomId);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandle_id();
        roomExistRequest.setSession_id(session_id);
        roomExistRequest.setHandle_id(handle_id);
        roomExistRequest.setBody(roomExistBody);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(roomExistRequest, nodeId).get();
        return response.getJSONObject("plugindata").getJSONObject("data").getBoolean("exists");
    }

    public String createRoom(String userId, CreateRoomBody createRoomBody) throws Exception {
        JanusRequest createRoomRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandle_id();
        createRoomRequest.setSession_id(session_id);
        createRoomRequest.setHandle_id(handle_id);
        createRoomRequest.setBody(createRoomBody);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(createRoomRequest, nodeId).get();
        return response.toString();
    }


    public String join(String userId, String targetUserId, JoinBody joinBody) throws Exception {
        JanusRequest joinRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id;
        if (targetUserId == null) {
            handle_id = participant.getHandle_id();
        } else {
            handle_id = participant.getHandleIds().get(targetUserId);
        }
        joinRequest.setSession_id(session_id);
        joinRequest.setHandle_id(handle_id);
        joinRequest.setBody(joinBody);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(joinRequest, nodeId).get();
        response.put("handle_id", handle_id);
        return response.toString();
    }

    public String sendOffer(String userId, Jsep jsep, ConfigureBody configureBody) throws Exception {
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandle_id();
        JanusRequest offerRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        offerRequest.setJsep(jsep);
        offerRequest.setBody(configureBody);
        offerRequest.setSession_id(session_id);
        offerRequest.setHandle_id(handle_id);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(offerRequest, nodeId).get();
        return response.toString();

    }

    public String sendCandidate(String userId, String targetUserId, String candidate) {
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id;
        if (userId.equals(targetUserId)) {
            handle_id = participant.getHandle_id();
        } else {
            handle_id = participant.getHandleIds().get(targetUserId);
        }
        JanusRequest candidateRequest = new JanusRequest(JanusNativeSignaling.TRICKLE);
        candidateRequest.setCandidate(candidate);
        candidateRequest.setSession_id(session_id);
        candidateRequest.setHandle_id(handle_id);
        int nodeId = janusUtil.getUserNode(userId);
        janusRequestSender.sendWithFuture(candidateRequest, nodeId);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("janus", "ack");
        return jsonObject.toString();
    }

    public String keepalive(String userId) {
        JanusRequest keepaliveRequest = new JanusRequest(JanusNativeSignaling.KEEP_A_LIVE);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        if (participant == null) {
            return null;
        }
        long session_id = participant.getSession_id();
        keepaliveRequest.setSession_id(session_id);
        int nodeId = janusUtil.getUserNode(userId);
        janusRequestSender.sendWithFuture(keepaliveRequest, nodeId);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("janus", "ack");
        return jsonObject.toString();
    }

    public String leave(String userId) throws Exception {
        JanusRequest leaveRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        if (participant == null) {
            return null;
        }
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandle_id();
        JanusRequestBody leaveRequestBody = new JanusRequestBody(JanusNativeSignaling.LEAVE);
        leaveRequest.setBody(leaveRequestBody);
        leaveRequest.setSession_id(session_id);
        leaveRequest.setHandle_id(handle_id);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(leaveRequest, nodeId).get();
        return response.toString();
    }

    public String destroy(String userId, int room) throws Exception {
        JanusRequest destroyRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandle_id();
        destroyRequest.setSession_id(session_id);
        destroyRequest.setHandle_id(handle_id);
        JanusRequestBody destroyRequestBody = new JanusRequestBody(JanusNativeSignaling.DESTROY);
        destroyRequestBody.setRoom(room);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(destroyRequest, nodeId).get();
        return response.toString();
    }

    public String configure(String userId, String targetUserId,ConfigureBody configureBody) {
        JanusRequest configureRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id;
        if (userId.equals(targetUserId)) {
            handle_id = participant.getHandle_id();
        } else {
            handle_id = participant.getHandleIds().get(targetUserId);
        }
        configureRequest.setBody(configureBody);
        configureRequest.setSession_id(session_id);
        configureRequest.setHandle_id(handle_id);
        int nodeId = janusUtil.getUserNode(userId);
        janusRequestSender.sendWithFuture(configureRequest, nodeId);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("janus", "ack");
        return jsonObject.toString();
    }

    private void rtpForward(String userId, String targetUserId, RtpForwardBody rtpForwardBody) throws Exception {
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandle_id();
        JanusRequest rtpForwardRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        rtpForwardRequest.setBody(rtpForwardBody);
        rtpForwardRequest.setSession_id(session_id);
        rtpForwardRequest.setHandle_id(handle_id);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(rtpForwardRequest, nodeId).get();
        addProducer(participant, targetUserId, rtpForwardBody, response);
        response.toString();
    }

    private void addProducer(Participant participant, String targetUserId, RtpForwardBody rtpForwardBody, JSONObject rtpForwardResponse) {
        Producer producer = new Producer();
        producer.setOriginHost(participant.getHost());
        producer.setHost(rtpForwardBody.getHost());
        producer.setStreamId(rtpForwardBody.getId());
        producer.setRoom(rtpForwardBody.getRoom());
        producer.setRtpForwardTo(targetUserId);
        producer.setAudio_port(rtpForwardBody.getAudio_port());
        producer.setVideo_port(rtpForwardBody.getVideo_port());
        producer.setId(rtpForwardBody.getPublisher_id());
        producer.setAudioStreamId(rtpForwardResponse.getJSONObject("plugindata").getJSONObject("data").getJSONObject("rtp_stream").getLong("audio_stream_id"));
        producer.setVideoStreamId(rtpForwardResponse.getJSONObject("plugindata").getJSONObject("data").getJSONObject("rtp_stream").getLong("video_stream_id"));
        producer.setVideo_rtcp_port(rtpForwardBody.getVideo_rtcp_port());
        janusUtil.putProducers(participant.getDisplay(), producer);
    }

    public void stopMyRtpForward(String userId) {
        if (!janusUtil.isProducersContains(userId)) {
            janusLogUtil.log("no RTP forwarder had to destroy.", this.getClass(), JanusLogUtil.INFO);
            return;
        }
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandle_id();
        int nodeId = participant.getNode();
        JanusRequest stopRtpForwardRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        stopRtpForwardRequest.setSession_id(session_id);
        stopRtpForwardRequest.setHandle_id(handle_id);
        RtpForwardBody rtpForwardBody = new RtpForwardBody(JanusNativeSignaling.STOP_RTP_FORWARD);
        List<Producer> producerList = janusUtil.getProducers(userId);
        for (Producer p : producerList) {
            rtpForwardBody.setRoom(p.getRoom());
            rtpForwardBody.setPublisher_id(p.getId());
            rtpForwardBody.setStream_id(p.getAudioStreamId());
            stopRtpForwardRequest.setBody(rtpForwardBody);
            janusRequestSender.sendWithFuture(stopRtpForwardRequest, nodeId);
            rtpForwardBody.setStream_id(p.getVideoStreamId());
            stopRtpForwardRequest.setBody(rtpForwardBody);
            janusRequestSender.sendWithFuture(stopRtpForwardRequest, nodeId);
        }
        janusUtil.removeProducers(userId);
    }

    public void stopOtherRtpForward(String rtpForwardFrom, String rtpForwardTo) {
        if (!janusUtil.isProducersContains(rtpForwardFrom)) {
            janusLogUtil.log("no RTP forwarder had to destroy.", this.getClass(), JanusLogUtil.INFO);
            return;
        }
        Participant participant = janusUtil.getParticipantByUserId(rtpForwardFrom);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandle_id();
        int nodeId = participant.getNode();
        JanusRequest stopRtpForwardRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        stopRtpForwardRequest.setSession_id(session_id);
        stopRtpForwardRequest.setHandle_id(handle_id);
        RtpForwardBody rtpForwardBody = new RtpForwardBody(JanusNativeSignaling.STOP_RTP_FORWARD);
        List<Producer> producerList = janusUtil.getProducers(rtpForwardFrom);
        List<Producer> producer = new ArrayList<>();
        for (Producer p : producerList) {
            if (p.getRtpForwardTo().equals(rtpForwardTo)) {
                producer.add(p);
                rtpForwardBody.setRoom(p.getRoom());
                rtpForwardBody.setPublisher_id(p.getId());
                rtpForwardBody.setStream_id(p.getAudioStreamId());
                stopRtpForwardRequest.setBody(rtpForwardBody);
                janusRequestSender.sendWithFuture(stopRtpForwardRequest, nodeId);
                rtpForwardBody.setStream_id(p.getVideoStreamId());
                stopRtpForwardRequest.setBody(rtpForwardBody);
                janusRequestSender.sendWithFuture(stopRtpForwardRequest, nodeId);
            }
        }
        if (producer.size() != 0) {
            producerList.removeAll(producer);
            if (producerList.size() == 0) {
                janusUtil.removeProducers(rtpForwardFrom);
            } else {
                janusUtil.putProducers(rtpForwardFrom, producerList);
            }
        }
    }

    public String sendRestartICE(String userId) throws Exception {
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long session_id = participant.getSession_id();
        long handle_id = participant.getHandle_id();
        JanusRequest configureRequest = new JanusRequest(JanusNativeSignaling.MESSAGE);
        ConfigureBody configureBody = new ConfigureBody();
        configureBody.setRestart(true);
        configureRequest.setBody(configureBody);
        configureRequest.setSession_id(session_id);
        configureRequest.setHandle_id(handle_id);
        int nodeId = janusUtil.getUserNode(userId);
        JSONObject response = janusRequestSender.sendWithFuture(configureRequest, nodeId).get();
        return response.toString();
    }

    public RtpForwardData createRtpForwardRequest(String userId) {
        RtpForwardData rtpForwardData = new RtpForwardData();
        rtpForwardData.setForwardFrom(userId);
        Random random = new Random();
        int id = random.nextInt(9999999) + 1000000;
        rtpForwardData.setId(id);
        Node node = janusUtil.getNodeInfo(janusUtil.getUserNode(userId));
        rtpForwardData.setClusterId(node.getClusterId());
        rtpForwardData.setHost(node.getHost());
        rtpForwardData.setInterHost(node.getInterHost());
        rtpForwardData.setAudio_port(janusUtil.getPort());
        rtpForwardData.setVideo_port(janusUtil.getPort());
        rtpForwardData.setVideo_port_2(janusUtil.getPort());
        rtpForwardData.setVideo_port_3(janusUtil.getPort());
        rtpForwardData.setVideo_rtcp_port(janusUtil.getPort());
        return rtpForwardData;
    }


    public void handleRtpForwardRequest(String rtpForwardFrom, String rtpForwardTo, RtpForwardData rtpForwardData) throws Exception {
        Participant participant = janusUtil.getParticipantByUserId(rtpForwardFrom);
        long publisher_id = participant.getPublisher_id();
        int roomId = Integer.parseInt(janusUtil.getUserRoom(rtpForwardFrom));
        RtpForwardBody rtpForwardBody = new RtpForwardBody();
        rtpForwardBody.setRoom(roomId);
        /**
         * 如果janus属于同于内外 则走内网转发
         */
        String host;
        String myHost;
        Node myNode = janusUtil.getNodeInfo(janusUtil.getUserNode(rtpForwardFrom));
        if (myNode.getClusterId().equals(rtpForwardData.getClusterId())) {
            myHost = myNode.getInterHost();
            host = rtpForwardData.getInterHost();
        } else {
            myHost = myNode.getHost();
            host = rtpForwardData.getHost();
        }
        participant.setHost(myHost);
        janusUtil.updateParticipant(rtpForwardFrom, participant);
        rtpForwardBody.setHost(host);
        rtpForwardBody.setAudio_port(rtpForwardData.getAudio_port());
        rtpForwardBody.setPublisher_id(publisher_id);
        rtpForwardBody.setVideo_rtcp_port(rtpForwardData.getVideo_rtcp_port());
        rtpForwardBody.setVideo_port(rtpForwardData.getVideo_port());
        rtpForwardBody.setVideo_port_2(rtpForwardData.getVideo_port_2());
        rtpForwardBody.setVideo_port_3(rtpForwardData.getVideo_port_3());
        rtpForward(rtpForwardFrom, rtpForwardTo, rtpForwardBody);
    }


    /**
     * 处理create_room,只有发起者会调用
     * handle create_room signal
     * only the room owner could send this signal
     *
     * @throws Exception
     */
    public String handleCreateRoom(String userId, CreateRoomBody createRoomBody) throws Exception {
        String roomId = String.valueOf(createRoomBody.getRoom());
        int nodeId = janusUtil.getFreeNode(roomId);
        janusUtil.putUserToNode(userId, nodeId);
        if (janusUtil.isUserRoomContainsKey(userId)) {
            String oldRoomId = janusUtil.getUserRoom(userId);
            janusLogUtil.log("用户:" + userId + "在房间:" + oldRoomId + "中", this.getClass(), JanusLogUtil.TRACE);
            if (!oldRoomId.equals(roomId)) {
                janusUtil.leaveRoom(userId);
                janusLogUtil.log(userId + "离开房间:" + oldRoomId, this.getClass(), JanusLogUtil.TRACE);
            }
        }
        janusUtil.putUserRoom(userId, roomId);
        long session_id = createSession(userId);
        janusLogUtil.log(userId + "创建session_id:" + session_id, this.getClass(), JanusLogUtil.TRACE);
        long handle_id = createHandle(userId, session_id, JanusPlugin.VIDEOROOM);
        janusLogUtil.log(userId + "创建handle_id:" + session_id + "插件:" + JanusPlugin.VIDEOROOM, this.getClass(), JanusLogUtil.TRACE);
        Participant participant = new Participant();
        participant.setSession_id(session_id);
        participant.setHandle_id(handle_id);
        participant.setDisplay(userId);
        participant.setNode(nodeId);
        participant.setInRoom(false);
        janusUtil.updateParticipant(userId, participant);
        String response = createRoom(userId, createRoomBody);
        janusLogUtil.log(userId + "创建房间:" + roomId, this.getClass(), JanusLogUtil.TRACE);
        return response;
    }

    /**
     * 处理join_pub
     * 把用户添加到某一个节点（根据节点分配策略）
     * 把用户添加到房间中
     * 进入房间（如果是分布式的的时候，房间不存在，则会在新服务器创建一个房间）
     * 并通知远端服务器的其他用户
     * 查询远端服务器的用户，返回到remotePublisher参数中
     * handle join_pub
     * 1.put user to a janus node (depends on the load distribution strategy)
     * 2.put user in the room
     * 3.join room (if in the distribution env, room may be not exist, In this situation, we will create a new room which use the same room id in the new janus node)
     * 4.notify the participant whose are in the remote server
     * 5.get the participant info and return as remotePublisher
     */
    public String handleJoinAndPub(String userId, JoinBody joinBody) throws Exception {
        String roomId = String.valueOf(joinBody.getRoom());
        String key = String.valueOf(roomId);
        RedissLockUtil.tryLock(key, TimeUnit.SECONDS, 10, 3);
        janusLogUtil.log(Thread.currentThread().getName() + "||" + key + "获取锁", this.getClass(), JanusLogUtil.TRACE);
        try {
            if (janusUtil.isUserRoomContainsKey(userId)) {
                String oldRoomId = janusUtil.getUserRoom(userId);
                if (!oldRoomId.equals(roomId)) {
                    janusUtil.leaveRoom(userId);
                    janusLogUtil.log(userId + "离开房间:" + oldRoomId, this.getClass(), JanusLogUtil.TRACE);
                }
            }
            janusLogUtil.log(userId + "加入房间:" + roomId, this.getClass(), JanusLogUtil.TRACE);
            int nodeId = 0;
            Participant participant = janusUtil.getParticipantByUserId(userId);
            long session_id;
            long handle_id;
            if (participant == null) {
                nodeId = janusUtil.putUserToNode(userId, janusUtil.getFreeNode(roomId));
                participant = new Participant();
                session_id = createSession(userId);
                janusLogUtil.log(userId + "创建session_id:" + session_id, this.getClass(), JanusLogUtil.TRACE);
                handle_id = createHandle(userId, session_id, JanusPlugin.VIDEOROOM);
                janusLogUtil.log(userId + "创建handle_id:" + session_id + "插件:" + JanusPlugin.VIDEOROOM, this.getClass(), JanusLogUtil.TRACE);
                participant.setSession_id(session_id);
                participant.setHandle_id(handle_id);
                participant.setDisplay(userId);
                participant.setNode(nodeId);
            }
            janusUtil.putUserRoom(userId, roomId);
            participant.setInRoom(true);
            joinBody.setId(participant.getPublisher_id());
            janusUtil.updateParticipant(userId, participant);
            notifyRemoteParticipantJoining(userId);
            if (!roomExist(userId, Integer.parseInt(roomId))) {
                CreateRoomBody createRoomBody = new CreateRoomBody();
                createRoomBody.setRoom(joinBody.getRoom());
                createRoomBody.setPublishers(9);
                createRoomBody.setIs_private(true);
                createRoomBody.setNotify_joining(true);
                createRoom(userId, createRoomBody);
                janusLogUtil.log(userId + "创建房间:" + roomId, this.getClass(), JanusLogUtil.TRACE);
            }
            joinBody.setPtype("publisher");
            String response = join(userId, null, joinBody);
            JSONObject resp = JSONObject.parseObject(response);
            JSONObject plugindata = resp.getJSONObject("plugindata");
            JSONObject data = plugindata.getJSONObject("data");
            JSONArray publishers = data.getJSONArray("publishers");
            if (publishers == null) {
                publishers = new JSONArray();
            }
            List<Participant> remotePublishers = janusUtil.getAVRoomByUserId(userId).getRemoteParticipant(nodeId);
            Publisher publisher;
            for (Participant remotePublisher : remotePublishers) {
                publisher = new Publisher();
                publisher.setId(remotePublisher.getPublisher_id());
                publisher.setDisplay(remotePublisher.getDisplay());
                publisher.setRemote(true);
                publishers.add(publisher);
            }
            data.put("publishers", publishers);
            plugindata.put("data", data);
            resp.put("plugindata", plugindata);
            response = resp.toJSONString();
            return response;
        } finally {
            RedissLockUtil.unlock(key);
            janusLogUtil.log(Thread.currentThread().getName() + "||" + key + "释放锁", this.getClass(), JanusLogUtil.TRACE);
        }
    }

    /**
     * 处理离开
     * 销毁所有rtp转发
     * 离开房间
     * 把用户从节点移除
     * 把用户从房间移除
     * handle leave
     * 1.destroy all rtp forwarder (if exist)
     * 2.leave room
     * 3.remove user from node
     * 4.remove user from room
     *
     * @param userId
     * @return
     * @throws Exception
     */
    public String handleLeave(String userId) throws Exception {
        String roomId = janusUtil.getUserRoom(userId);
        String key = roomId;
        RedissLockUtil.tryLock(key, TimeUnit.SECONDS, 10, 3);
        janusLogUtil.log(Thread.currentThread().getName() + "||" + key + "获取锁", this.getClass(), JanusLogUtil.TRACE);
        try {
            stopMyRtpForward(userId);
            janusLogUtil.log(userId + "暂停所有rtp转发", this.getClass(), JanusLogUtil.TRACE);
            if (!janusUtil.getIsShareStream()) {
                List<Subscriber> subscribers = janusUtil.getSubscribers(userId);
                for (Subscriber subscriber : subscribers) {
                    stopOtherRtpForward(subscriber.getSubscriberFrom(), userId);
                    janusLogUtil.log(subscriber.getSubscriberFrom() + "暂停所有rtp转发", this.getClass(), JanusLogUtil.TRACE);
                }
                destroyMyStream(userId);
                janusLogUtil.log(userId + "暂停所有stream接收", this.getClass(), JanusLogUtil.TRACE);
            }
            String response = leave(userId);
            notifyRemoteParticipantLeaving(userId);
            janusLogUtil.log(userId + "离开janus房间", this.getClass(), JanusLogUtil.TRACE);
            janusUtil.removeUserFromNode(userId);
            janusLogUtil.log(userId + "离开节点", this.getClass(), JanusLogUtil.TRACE);
            janusUtil.leaveRoom(userId);
            janusLogUtil.log(userId + "离开房间", this.getClass(), JanusLogUtil.TRACE);
            return response;
        } finally {
            RedissLockUtil.unlock(key);
            janusLogUtil.log(Thread.currentThread().getName() + "||" + key + "释放锁", this.getClass(), JanusLogUtil.TRACE);
        }
    }

    /**
     * 处理join_sub订阅
     * 如果订阅的用户不在同一个节点上，
     * 则使用rtp流转发
     * 如果已经为订阅的用户在当前节点创建了一个流接收器
     * 则不需要重复创建
     * 订阅的时候watch参数选择该streamId即可
     * handle join_sub
     * 1.if the user who we want to subscribe is in the different node,
     * in this situation, we could use the rtp forwarder to transport the rtp data from one janus node to the other janus node,
     * but if we already create a rtp forwarder for this user, We can use it directly.
     *
     * @param userId
     * @return
     * @throws Exception
     */
    public String handleJoinAndSub(String userId, String targetUserId, JoinBody joinBody) throws Exception {
        Participant participant = janusUtil.getParticipantByUserId(userId);
        boolean differentNode = false;
        long handle_id;
        int nodeId = janusUtil.getUserNode(userId);
        if (nodeId != janusUtil.getUserNode(targetUserId)) {
            differentNode = true;
            /**
             * 如果订阅的用户不在同一个节点上，
             */
            handle_id = createHandle(userId, participant.getSession_id(), JanusPlugin.STREAMING);
            janusLogUtil.log(userId + "创建handle_id:" + handle_id + "插件:" + JanusPlugin.STREAMING, this.getClass(), JanusLogUtil.TRACE);
        } else {
            handle_id = createHandle(userId, participant.getSession_id(), JanusPlugin.VIDEOROOM);
            janusLogUtil.log(userId + "创建handle_id:" + handle_id + "插件:" + JanusPlugin.VIDEOROOM, this.getClass(), JanusLogUtil.TRACE);
        }
        HashMap<String, Long> handleIds = participant.getHandleIds();
        handleIds.put(targetUserId, handle_id);
        participant.setHandleIds(handleIds);
        janusUtil.updateParticipant(userId, participant);
        /**
         * 如果订阅的用户不在同一个节点上，
         */
        String response;
        if (differentNode) {
            int streamId = 0;
            if (janusUtil.getIsShareStream()) {
                /**
                 * 如果已经为订阅的用户在当前节点创建了一个流接收器
                 */
                int roomId = Integer.parseInt(janusUtil.getUserRoom(userId));
                streamId = janusUtil.getStreamId(targetUserId, janusUtil.getNodeInfo(nodeId).getHost(), roomId);
            }
            if (streamId == 0) {
                RtpForwardData rtpForwardData = createRtpForwardRequest(userId);
                //创建接收流
                streamId = rtpForwardData.getId();
                createStream(userId, targetUserId, rtpForwardData);
                handleRtpForwardRequest(targetUserId, userId, rtpForwardData);
            }
            //选择观看流
            response = watch(userId, targetUserId, streamId);
        } else {
            joinBody.setRoom(janusUtil.getUserRoom(userId));
            joinBody.setPtype("subscriber");
            joinBody.setFeed(janusUtil.getParticipantByUserId(targetUserId).getPublisher_id());
            response = join(userId, targetUserId, joinBody);
        }
        JSONObject resp = JSONObject.parseObject(response);
        JSONObject plugindata = resp.getJSONObject("plugindata");
        JSONObject data = plugindata.getJSONObject("data");
        if (!data.containsKey("display")) {
            data.put("display", targetUserId);
            plugindata.put("data", data);
            resp.put("plugindata", plugindata);
            response = resp.toJSONString();
        }
        return response;
    }

    public void notifyRemoteParticipantJoining(String userId) {
        Room room = janusUtil.getAVRoomByUserId(userId);
        Participant user = room.getParticipantHashMap().get(userId);
        List<Participant> remoteParticipant = room.getRemoteParticipant(janusUtil.getUserNode(userId));
        for (Participant participant : remoteParticipant) {
            Event remoteEvent = new Event();
            remoteEvent.setSender(participant.getHandle_id());
            remoteEvent.setSession_id(participant.getSession_id());
            Plugindata plugindata = new Plugindata();
            Data data = new Data();
            Publisher publisher = new Publisher();
            publisher.setDisplay(userId);
            publisher.setId(user.getPublisher_id());
            publisher.setRemote(true);
            List<Publisher> publishers = new ArrayList<>();
            publishers.add(publisher);
            data.setPublishers(publishers);
            janusLogUtil.log("通知用户:" + participant.getDisplay() + " " + userId + "加入", this.getClass(), JanusLogUtil.TRACE);
            plugindata.setData(data);
            remoteEvent.setPlugindata(plugindata);
            imSender.sendEvent(participant.getDisplay(), remoteEvent);
            webSocketSender.sendEvent(participant.getDisplay(), remoteEvent);
        }
    }

    /**
     * 通知房间内所有用户
     *
     * @param userId
     */
    public void notifyAllParticipantLeaving(String userId) {
        Room room = janusUtil.getAVRoomByUserId(userId);
        Participant user = room.getParticipantHashMap().get(userId);
        List<Participant> remoteParticipant = room.getParticipantList();
        for (Participant participant : remoteParticipant) {
            if(userId.equals(participant.getDisplay())){
                //不通知自己
                continue;
            }
            Event remoteEvent = new Event();
            remoteEvent.setSender(participant.getHandle_id());
            remoteEvent.setSession_id(participant.getSession_id());
            Plugindata plugindata = new Plugindata();
            Data data = new Data();
            data.setLeaving(user.getDisplay());
            janusLogUtil.log("通知用户:"+participant.getDisplay()+" "+userId+"离开",this.getClass(),JanusLogUtil.TRACE);
            plugindata.setData(data);
            remoteEvent.setPlugindata(plugindata);
            imSender.sendEvent(participant.getDisplay(), remoteEvent);
            webSocketSender.sendEvent(participant.getDisplay(), remoteEvent);
        }
    }

    /**
     * 通知房间其他节点用户
     * @param userId
     */
    public void notifyRemoteParticipantLeaving(String userId) {
        Room room = janusUtil.getAVRoomByUserId(userId);
        Participant user = room.getParticipantHashMap().get(userId);
        List<Participant> remoteParticipant = room.getRemoteParticipant(userId);
        for (Participant participant : remoteParticipant) {
            if(userId.equals(participant.getDisplay())){
                //不通知自己
                continue;
            }
            Event remoteEvent = new Event();
            remoteEvent.setSender(participant.getHandle_id());
            remoteEvent.setSession_id(participant.getSession_id());
            Plugindata plugindata = new Plugindata();
            Data data = new Data();
            data.setLeaving(user.getDisplay());
            janusLogUtil.log("通知用户:"+participant.getDisplay()+" "+userId+"离开",this.getClass(),JanusLogUtil.TRACE);
            plugindata.setData(data);
            remoteEvent.setPlugindata(plugindata);
            imSender.sendEvent(participant.getDisplay(), remoteEvent);
            webSocketSender.sendEvent(participant.getDisplay(), remoteEvent);
        }
    }

    public String startLive(String userId) throws Exception{
        Participant participant = janusUtil.getParticipantByUserId(userId);
        long handle_id = createHandle(userId, participant.getSession_id(), JanusPlugin.STREAMING);
        janusLogUtil.log(userId + "创建handle_id:" + handle_id + "插件:" + JanusPlugin.STREAMING, this.getClass(), JanusLogUtil.TRACE);
        HashMap<String, Long> handleIds = participant.getHandleIds();
        handleIds.put(userId, handle_id);
        participant.setHandleIds(handleIds);
        janusUtil.updateParticipant(userId, participant);
        RtpForwardData rtpForwardData = createRtpForwardRequest(userId);
        //创建接收流
        createStream(userId, rtpForwardData);
        handleRtpForwardRequest(userId, userId, rtpForwardData);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("janus", "ack");
        return jsonObject.toString();
    }

    public String stopLive(String userId) throws Exception{
        stopMyRtpForward(userId);
        janusLogUtil.log(userId + "暂停所有rtp转发", this.getClass(), JanusLogUtil.TRACE);
        if (!janusUtil.getIsShareStream()) {
            List<Subscriber> subscribers = janusUtil.getSubscribers(userId);
            for (Subscriber subscriber : subscribers) {
                stopOtherRtpForward(subscriber.getSubscriberFrom(), userId);
                janusLogUtil.log(subscriber.getSubscriberFrom() + "暂停所有rtp转发", this.getClass(), JanusLogUtil.TRACE);
            }
            destroyMyStream(userId);
            janusLogUtil.log(userId + "暂停所有stream接收", this.getClass(), JanusLogUtil.TRACE);
        }
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("janus", "ack");
        return jsonObject.toString();
    }

    public String watchLive(String userId,int roomId)throws Exception{
        Participant participant = new Participant();
        int nodeId = janusUtil.getRoomNodeId(String.valueOf(roomId));
        janusUtil.putUserToNode(userId,nodeId);
        long session_id = createSession(userId);
        long handle_id = createHandle(userId, session_id, JanusPlugin.STREAMING);
        janusLogUtil.log(userId + "创建handle_id:" + handle_id + "插件:" + JanusPlugin.STREAMING, this.getClass(), JanusLogUtil.TRACE);
        HashMap<String, Long> handleIds = participant.getHandleIds();
        handleIds.put(userId, handle_id);
        participant.setDisplay(userId);
        participant.setSession_id(session_id);
        participant.setHandleIds(handleIds);
        participant.setHandle_id(handle_id);
        Random random = new Random();
        janusUtil.putUserRoom(userId,""+(random.nextInt(99999)+10000));
        janusUtil.updateParticipant(userId, participant);
        return watch(userId,userId,roomId);
    }
}
